package com.example.toy;

import com.example.toy.core.ToySystemFactoryProducer;
import com.example.toy.core.archive.ToyArchiveSystem;
import com.example.toy.core.conf.ToyConfigs;
import com.example.toy.core.conf.ToySystemConfig;
import com.example.toy.core.conf.ToyTableConfig;
import com.example.toy.core.exception.NotExistsException;
import com.example.toy.core.system.AbstractToySystem;
import com.example.toy.core.table.TableIdentifier;
import com.example.toy.core.table.ToyTable;
import com.example.toy.core.version.ToyVersion;
import lombok.SneakyThrows;
import org.apache.hadoop.fs.Path;

import java.lang.reflect.Constructor;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;

public interface ToySystem extends ToyArchiveSystem {

  /**
   * 查询所有的表
   *
   * @param catalog
   * @param database
   * @param tablePattern
   * @return
   */
  List<TableIdentifier> list(String catalog, String database, String tablePattern);

  /**
   * 获取表的基本路径
   *
   * @param tableIdentifier
   * @return
   */
  Path getToyTablePath(TableIdentifier tableIdentifier) throws NotExistsException;

  /**
   * 根据表ID和版本号获取表
   *
   * @param tableIdentifier
   * @param version
   * @return
   */
  ToyTable getToyTable(TableIdentifier tableIdentifier, ToyVersion version);

  /**
   * 获取最新版本的可读数据的表
   *
   * @param tableIdentifier
   * @return
   */
  default ToyTable getToyTable(TableIdentifier tableIdentifier) {
    return getToyTable(tableIdentifier, ToyVersion.LATEST_VERSION);
  }

  /**
   * 根据元数据初始化表
   *
   * @param config
   * @return
   */
  ToyTable create(TableIdentifier tableIdentifier, ToyTableConfig config);

  /**
   * 删除表
   *
   * @param tableIdentifier
   */
  default void drop(TableIdentifier tableIdentifier) {
    getToyTable(tableIdentifier).drop();
  }



  /**
   * 获取toy超级工厂
   *
   * @return
   */
  ToySystemFactoryProducer getSystemFactoryProducer();



  static Builder builder() {
    return new Builder();
  }

  class Builder {
    private final ToySystemConfig config;

    public Builder() {
      this.config = new ToySystemConfig();
    }

    public Builder name(String name) {
      config.setConf(ToyConfigs.TOY_SYSTEM_NAME.getKey(), name);
      return this;
    }

    public Builder type(String type) {
      config.setConf(ToyConfigs.TOY_SYSTEM_TYPE.getKey(), type);
      return this;
    }

    public Builder conf(String key, String value) {
      config.setConf(key, value);
      return this;
    }

    public Builder conf(String key, Integer value) {
      config.setConf(key, value == null ? null : value.toString());
      return this;
    }

    public Builder conf(String key, Long value) {
      config.setConf(key, value == null ? null : value.toString());
      return this;
    }

    public Builder conf(String key, Double value) {
      config.setConf(key, value == null ? null : value.toString());
      return this;
    }

    public Builder conf(String key, Float value) {
      config.setConf(key, value == null ? null : value.toString());
      return this;
    }

    public Builder conf(String key, Boolean value) {
      config.setConf(key, value == null ? null : value.toString());
      return this;
    }

    public Builder conf(Map<String, String> configs) {
      configs.forEach(
          (key, value) -> {
            config.setConf(key, value);
          });
      return this;
    }

    public Builder conf(ToySystemConfig configs) {
      Properties properties = configs.getProperties();
      Set<String> strings = properties.stringPropertyNames();
      for (String string : strings) {
        config.setConf(string, properties.getProperty(string));
      }
      return this;
    }

    public ToySystem build() {
      return getOrCreate(this.config);
    }
  }

  @SneakyThrows
  static ToySystem getOrCreate(ToySystemConfig config) {
    config.setIfMissing(ToyConfigs.TOY_SYSTEM_NAME.getKey(), UUID.randomUUID().toString());
    config.setIfMissing(ToyConfigs.TOY_SYSTEM_TYPE.getKey(), ToySystemConfig.FSOFFLINE_SHORT_NAME);
    config.setIfMissing(ToySystemConfig.FSOFFLINE_PATH_KEY, ".");

    String systemName = config.getSystemName();
    ToySystem toySystem = ToySystemCache.getToySystem(systemName);
    if (toySystem != null) {
      return toySystem;
    } else {
      Class<? extends AbstractToySystem> clz = config.getSystemType();
      Constructor<? extends AbstractToySystem> constructor = clz.getConstructor(ToySystemConfig.class);
      constructor.setAccessible(true);
      AbstractToySystem system = constructor.newInstance(config);
      return system;
    }
  }
}
